package com.taxi.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;


/**
 * kafka 生产者
 */
@Component
@Slf4j
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    /**
     * 异步发送
     */
    public void sendMessage(String topic, String message) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                int partition = result.getRecordMetadata().partition();
                long offset = result.getRecordMetadata().offset();
                log.info("Message sent successfully to topic: {}, partition: {}, offset: {}",topic ,partition ,offset);
            }
            @Override
            public void onFailure(Throwable e) {
                log.error("Failed to send message: {}", message);
            }
        });
    }

    /**
     * 同步发送
     */
//    public void sendMessage(String message) {
//        try {
//            SendResult<String, String> result = kafkaTemplate.send("travel-topic", message).get();
//            String topic = result.getRecordMetadata().topic();
//            int partition = result.getRecordMetadata().partition();
//            long offset = result.getRecordMetadata().offset();
//            log.info("Message sent successfully to topic: {}, partition: {}, offset: {}", topic, partition, offset);
//        } catch (ExecutionException | InterruptedException e) {
//            // 处理异常
//        }
//    }
}
